/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements...
 */

package org.apache.flink.connector.elasticsearch.sink;

import org.apache.flink.connector.base.sink.writer.AsyncSinkWriterStateSerializer;

import java.io.DataInputStream;
import java.io.DataOutputStream;

/**
 * State serializer for AsyncSinkWriter that delegates per-request (de)serialization to
 * OperationSerializer bound to the task thread's TCCL. This guarantees co.elastic.* visibility
 * at restore time in distributed clusters.
 */
public class Elasticsearch8AsyncSinkSerializer extends AsyncSinkWriterStateSerializer<Operation> {

    // 缓存一个绑定了“当前线程 TCCL”的 OperationSerializer，整个（反）序列化周期统一使用
    private final OperationSerializer opSer =
            new OperationSerializer(Thread.currentThread().getContextClassLoader());

    @Override
    protected void serializeRequestToStream(Operation request, DataOutputStream out) {
        opSer.serialize(request, out);
    }

    @Override
    protected Operation deserializeRequestFromStream(long requestSize, DataInputStream in) {
        return opSer.deserialize(requestSize, in);
    }

    @Override
    public int getVersion() {
        return 1;
    }
}
